package com.group1.realtime.app.dwd.db;

import com.group1.realtime.app.BaseSQLApp;
import com.group1.realtime.common.Constant;
import com.group1.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DWD_TradeCartAdd extends BaseSQLApp {

    public static void main(String[] args) {
        new DWD_TradeCartAdd().init(
                2021,
                2,
                "DWD_TradeCartAdd"
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {

        //读取ods_db数据
        readOdsDb(tEnv, "DWD_TradeCartAdd");

        //从ods_db中过滤出加购数据
        Table cartInfo = tEnv.sqlQuery("select " +
                " `data`['id'] id, " +
                " `data`['user_id'] user_id, " +
                " `data`['course_id'] course_id, " +
                " `data`['cart_price'] cart_price, " +
                " `data`['session_id'] source_type_code, " +
                " ts " +
                "from ods_db " +
                "where `database`='edu' " +
                "and `table`='cart_info' " +
                "and `type`='insert' " );

        tEnv.executeSql("create table dwd_trade_cart_add(" +
                "id string," +
                "user_id string," +
                "course_id string," +
                "cart_price string," +
                "session_id string," +
                "ts  String" +
                ")" + SQLUtil.getKafkaSinkDDL(Constant.TOPIC_DWD_TRADE_CART_ADD));

        cartInfo.executeInsert("dwd_trade_cart_add");
    }
}
